package ma.zhiqiang.dataline.core.convert;

import cn.hutool.json.JSONArray;
import cn.hutool.json.JSONUtil;
import lombok.AllArgsConstructor;
import lombok.NoArgsConstructor;
import ma.zhiqiang.dataline.core.dto.Column;
import ma.zhiqiang.dataline.core.dto.Schema;
import ma.zhiqiang.dataline.core.dto.SourceRecord;
import ma.zhiqiang.dataline.util.ClassTypeUtil;

import java.util.List;
import java.util.Properties;

public class RecordConvertCsv implements RecordConvert{
    Properties properties;
    List<Schema> schemas;       //csv数据字段定义
    String delimiter;           //csv分隔符
    public RecordConvertCsv(Properties properties) {
        this.properties = properties;
        this.schemas = JSONUtil.toList((JSONArray) properties.get("schema"),Schema.class);
        this.delimiter = properties.getProperty("delimiter",",");
    }

    @Override
    public SourceRecord convert(Object record) {
        String recordString = (String) record;
        String[] values = recordString.split(delimiter);
        if(null==values || values.length==0){return null;}
        if(values.length < schemas.size()){ throw new RuntimeException("schema集合长度应小于等于数据字段集合长度");}
        Boolean indexOutOfBounds = schemas.stream().anyMatch(schema -> schema.getIndex()>=values.length);
        if(indexOutOfBounds){throw new RuntimeException("schema配置的index，出现下标越界");}
        SourceRecord sourceRecord = new SourceRecord(schemas.size());
        schemas.stream().forEach(schema -> {
            String alias = schema.getAlias();
            Class type = ClassTypeUtil.getType(schema.getType());
            String value = values[schema.getIndex()];
            sourceRecord.addColumn(new Column(alias,type,value),schema.getHandler());
        });
        return sourceRecord;
    }
}
